Amazon Connectで複数の電話先に順番に掛けるシンプルな方法

Amazon Connectから順番に電話を掛ける方法を紹介します。

#Amazon Connect

#AWS

荒平祐次(arap)

2023.03.15

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、AWS事業本部の荒平(@0Air)です。

Amazon Connectを用いて、複数の電話先に順番に電話を掛けるシンプルな方法をご紹介します。
ユースケースとしては、保守などで担当者が固定されていて、CloudWatchなどをトリガーにして架電することを想定しています。

概要

登場する主なサービス

本エントリで紹介する構成に主に登場するサービスは以下です。

メリット

フラグ管理にDynamoDBやS3などを使わず、ConnectとLambdaのみを使用するので、構成がシンプル
電話番号を記載したテキストファイルやCSVファイルを管理しなくてよい

デメリット

レスポンスを待機するため、Lambdaの起動時間が長くなる (≒料金が増える可能性がある)

電話のフロー

実装したいフローのイメージです。
仮に担当者を3人として、A→B→C→A→...→タイムアウトとなるようにします。

作ってみた

Amazon Connect

インスタンスの作成などの手順はこちらの記事をご覧ください。

コンタクトフローの作成
今回は、以下のようなコンタクトフローを作成しました。
ID情報の取得
次の手順で使用するIDを取得してメモしておきます。

① 「Amazon ConnectのインスタンスID」は、Amazon Connectのコンソール画面に記載があります。

② 「コンタクトフローID」は、使用するコンタクトフロー画面の左下にARNが記載してあります。

Lambda

Lambdaの環境変数を使用するため、以下の情報を入力して保存します。

SOURCE_PHONE_NUMBER
- 電話を掛ける番号を記載します（ 発信側 ）
- インスタンスで使用する電話番号は、Amazon Connectサービス画面にて取得できます
TARGET_PHONE_NUMBER
- 電話を受ける番号を記載します（ 受信側 ）
- 電話番号については国番号(日本の場合+81)から始め、複数ある場合はカンマ(,)で区切ります
INSTANCE_ID
- Amazon ConnectのインスタンスIDを記載します
CONTACT_FLOW_ID
- 使用するAmazon ConnectのコンタクトフローIDを記載します

また、以下のコードを保存します。

import json
import boto3
import os
import time

SOURCE_PHONE_NUMBER = os.getenv('SOURCE_PHONE_NUMBER')
TARGET_PHONE_NUMBER = os.getenv('TARGET_PHONE_NUMBER')
INSTANCE_ID = os.getenv('INSTANCE_ID')
CONTACT_FLOW_ID = os.getenv('CONTACT_FLOW_ID')
Telnumbers = ""

def lambda_handler(event, context):
    global Telnumbers
    Telnumbers = os.environ[TARGET_PHONE_NUMBER].split(',')
    connect_call()

# Amazon Connectからの電話発信
def connect_call():
    call_status = "unanswered"
    print("架電対象：" + str(Telnumbers))
    for Telnumber in Telnumbers:
        if call_status=="unanswered":
            client = boto3.client('connect', region_name='ap-northeast-1')
            call = client.start_outbound_voice_contact(
                        DestinationPhoneNumber= Telnumber,
                        InstanceId=INSTANCE_ID,
                        ContactFlowId=CONTACT_FLOW_ID,
                        SourcePhoneNumber=SOURCE_PHONE_NUMBER,
                        Attributes = {
                            'call_status': call_status
                        }
                    )
            # 電話が取られているか判定
            time.sleep(10) # StartOutboundVoiceContactのタイムアウトは60秒
            contact_id = call['ContactId']
            attributes = client.get_contact_attributes(
                            InstanceId=INSTANCE_ID,
                            InitialContactId=contact_id
                        )
            call_status = attributes['Attributes']['call_status']
            print(json.dumps(attributes))
            print(Telnumber, call_status)

後半にて、call_statusの値を確認することで、誰かが電話を取った後に架電しない仕組みとしました。
コンタクトフロー内で「コンタクト属性の設定」を用いることで、フローが開始した（≒電話が繋がった）際に変数を再代入しています。